package com.tarena.luban.test.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 消费端 监听队列消费消息
 */
@Slf4j
public class MyConsumer01 {


    //3. 提供while true 禁止线程关闭
    @Test
    public void consumer01() throws Exception {
    //1. 构建消费者对象,连接nameserver创建长链接
        //创建一个默认消费者对象
        DefaultMQPushConsumer consumer= new DefaultMQPushConsumer();
        //给消费者提供必要属性
        consumer.setNamesrvAddr("localhost:9876");
        //消费者分组
        consumer.setConsumerGroup("consumer-group-02");
        //告诉消费者,你要监听绑定队列是谁,绑定主题
        consumer.subscribe("topic-01","*");
    //2. 执行api开始监听队列
        //提供给这个consumer一个监听器 listener
        consumer.setMessageListener(new MessageListenerConcurrently() {
            //每次拿到一个消息,就调用一下consumeMessage的方法
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //list就是封装了消息数据的对象,每次基本上list元素只有1个
                MessageExt messageExt = list.get(0);
                byte[] body = messageExt.getBody();
                //转化string
                String msg=new String(body, StandardCharsets.UTF_8);
                log.info("消费者分组:{},消费者01,消息数据:{}",consumer.getConsumerGroup(),msg);
                //根据消费消息的结果. success返回结果,会导致这个消费者分组的所有消费者
                //不可能再次消费这个消息
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    //3. 开启消费者长连接 线程不停止
        consumer.start();
        while(true);
    }
}
